package com.uuj.project.utils;

import java.util.HashMap;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.subjects.PublishSubject;
import io.reactivex.subjects.Subject;

/**
 * ProjectName:    ok_rxjava_retrofit
 * Package:        com.uuj.project.utils
 * ClassName:      RxBusUtil
 * Description:    描述
 * Author:         张继
 * CreateDate:     2019/9/6 14:25
 * UpdateUser:     更新者
 * UpdateDate:     2019/9/6 14:25
 * UpdateRemark:   更新说明
 * Version:        1.0
 */
public final class RxBusUtil {

    private HashMap<String, CompositeDisposable> mSubscriptionMap;
    private final Subject<Object> mSubject;//背压测试

    private static volatile RxBusUtil util = null;

    /**
     * 单例模式
     *
     * @return 返回当前实例对象
     */
    public static RxBusUtil getInstance() {
        if (util == null) {
            synchronized (RxBusUtil.class) {
                if (util == null) {
                    util = new RxBusUtil();
                }
            }
        }
        return util;
    }

    private RxBusUtil() {
        mSubject = PublishSubject.create().toSerialized();
    }

    public void post(Object obj) {
        if (hasObservers()) {
            mSubject.onNext(obj);
        }
    }

    /**
     * 返回指定类型的带背压的Flowable实例
     *
     * @param type 类型
     * @return 返回 Flowable 实例对象
     */
    private <T> Flowable<T> getObservable(Class<T> type) {
        return mSubject.toFlowable(BackpressureStrategy.BUFFER)
                .ofType(type);
    }

    /**
     * 一个默认的订阅方法
     *
     * @param type  类型
     * @param next  接受信息
     * @param error 错误信息
     * @return 返回 Disposable
     */
    public <T> Disposable doSubscribe(Class<T> type, Consumer<T> next, Consumer<Throwable> error) {

        return getObservable(type)
                .subscribeOn(Schedulers.io())
                .unsubscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(next, error);
    }

    /**
     * 是否已有观察者订阅
     *
     * @return true 有观察者 False无观察者
     */
    private boolean hasObservers() {
        return mSubject.hasObservers();
    }

    /**
     * 保存订阅后的disposable
     *
     * @param tClass     class
     * @param disposable 实例对象
     */
    public <T> void addSubscription(Class<T> tClass, Disposable disposable) {
        if (mSubscriptionMap == null) {
            mSubscriptionMap = new HashMap<>();
        }
        String key = tClass.getName();
        if (mSubscriptionMap.get(key) != null) {
            mSubscriptionMap.get(key).add(disposable);
        } else {
            //一次性容器,可以持有多个并提供 添加和移除。
            CompositeDisposable disposables = new CompositeDisposable();
            disposables.add(disposable);
            mSubscriptionMap.put(key, disposables);
        }
    }

    /**
     * 取消订阅
     *
     * @param tClass 类型
     */
    public <T> void unSubscribe(Class<T> tClass) {
        if (mSubscriptionMap == null) {
            return;
        }

        String key = tClass.getName();
        if (!mSubscriptionMap.containsKey(key)) {
            return;
        }
        if (mSubscriptionMap.get(key) != null) {
            mSubscriptionMap.get(key).dispose();
        }

        mSubscriptionMap.remove(key);
    }
}
